{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(1)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "pd.set_option('display.max_colwidth', 1000)\n",
    "pd.set_option('display.html.use_mathjax', False)\n",
    "\n",
    "DATA_DIR = ROOT_DIR + \"news_toutiao\" + os.sep\n",
    "\n",
    "ORIGIN_TRAIN_FILE = \"toutiao_cat_data.txt\";\n",
    "FIELD_DELIMITER = \"_!_\";\n",
    "\n",
    "SNN_MODEL_FILE = \"snn_model.ak\";\n",
    "APPROX_SNN_MODEL_FILE = \"approx_snn_model.ak\";\n",
    "LDA_MODEL_FILE = \"lda_model.ak\";\n",
    "LDA_PWZ_FILE = \"lda_pwz.ak\";\n",
    "\n",
    "SCHEMA_STRING = \"id string, category_code int, category_name string,\"\\\n",
    "                + \" news_title string, keywords string\";\n",
    "\n",
    "TXT_COL_NAME = \"news_title\";\n",
    "LABEL_COL_NAME = \"category_name\";\n",
    "PREDICTION_COL_NAME = \"pred\";\n",
    "\n",
    "def getSource() :\n",
    "    return CsvSourceBatchOp()\\\n",
    "            .setFilePath(DATA_DIR + ORIGIN_TRAIN_FILE)\\\n",
    "            .setSchemaStr(SCHEMA_STRING)\\\n",
    "            .setFieldDelimiter(FIELD_DELIMITER);\n",
    "\n",
    "def getStreamSource() :\n",
    "    return CsvSourceStreamOp()\\\n",
    "            .setFilePath(DATA_DIR + ORIGIN_TRAIN_FILE)\\\n",
    "            .setSchemaStr(SCHEMA_STRING)\\\n",
    "            .setFieldDelimiter(FIELD_DELIMITER);\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1\n",
    "\n",
    "getSource()\\\n",
    "    .lazyPrint(10)\\\n",
    "    .lazyPrintStatistics();\n",
    "\n",
    "getSource()\\\n",
    "    .groupBy(\"category_code, category_name\", \n",
    "             \"category_code, category_name, COUNT(category_name) AS cnt\")\\\n",
    "    .orderBy(\"category_code\", 100)\\\n",
    "    .lazyPrint(-1);\n",
    "\n",
    "BatchOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_1\n",
    "\n",
    "df = pd.DataFrame(\n",
    "    [\n",
    "        \"大家好！我在学习、使用Alink。\",\n",
    "        \"【流式计算和批式计算】、(Alink)\",\n",
    "        \"《人工智能》，“机器学习”？2020\"\n",
    "    ]\n",
    ") \n",
    "source = BatchOperator.fromDataframe(df, schemaStr='sentence string')\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        SegmentBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"words\")\n",
    "    )\\\n",
    "    .print();\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        SegmentBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"words\")\\\n",
    "            .setUserDefinedDict([\"流式计算\", \"机器学习\"])\n",
    "    )\\\n",
    "    .print();\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        SegmentBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"words\")\\\n",
    "            .setUserDefinedDict([\"流式计算\", \"机器学习\"])\n",
    "    )\\\n",
    "    .link(\n",
    "        StopWordsRemoverBatchOp()\\\n",
    "            .setSelectedCol(\"words\")\\\n",
    "            .setOutputCol(\"left_words\")\n",
    "    )\\\n",
    "    .print();\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        SegmentBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"words\")\\\n",
    "            .setUserDefinedDict([\"流式计算\", \"机器学习\"])\n",
    "    )\\\n",
    "    .link(\n",
    "        StopWordsRemoverBatchOp()\\\n",
    "            .setSelectedCol(\"words\")\\\n",
    "            .setOutputCol(\"left_words\")\\\n",
    "            .setStopWords([\"计算\", \"2020\"])\n",
    "    )\\\n",
    "    .print();\n",
    "\n",
    "getSource()\\\n",
    "    .select(\"news_title\")\\\n",
    "    .link(\n",
    "        SegmentBatchOp()\\\n",
    "            .setSelectedCol(\"news_title\")\\\n",
    "            .setOutputCol(\"segmented_title\")\n",
    "    )\\\n",
    "    .firstN(10)\\\n",
    "    .print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2_2\n",
    "\n",
    "df = pd.DataFrame(\n",
    "    [\n",
    "        \"Hello!      This is Alink!\",\n",
    "        \"Flink,Alink..AI#ML@2020\"\n",
    "    ]\n",
    ") \n",
    "source = BatchOperator.fromDataframe(df, schemaStr='sentence string')\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        TokenizerBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"tokens\")\n",
    "    )\\\n",
    "    .link(\n",
    "        RegexTokenizerBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"regex_tokens\")\n",
    "    )\\\n",
    "    .lazyPrint(-1);\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        RegexTokenizerBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"tokens_1\")\\\n",
    "            .setPattern(\"\\\\W+\")\n",
    "    )\\\n",
    "    .link(\n",
    "        RegexTokenizerBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"tokens_2\")\\\n",
    "            .setGaps(False)\\\n",
    "            .setPattern(\"\\\\w+\")\n",
    "    )\\\n",
    "    .lazyPrint(-1);\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        RegexTokenizerBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"tokens_1\")\\\n",
    "            .setPattern(\"\\\\W+\")\n",
    "    )\\\n",
    "    .link(\n",
    "        RegexTokenizerBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"tokens_2\")\\\n",
    "            .setPattern(\"\\\\W+\")\\\n",
    "            .setToLowerCase(False)\n",
    "    )\\\n",
    "    .lazyPrint(-1);\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        RegexTokenizerBatchOp()\\\n",
    "            .setSelectedCol(\"sentence\")\\\n",
    "            .setOutputCol(\"tokens\")\\\n",
    "            .setPattern(\"\\\\W+\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StopWordsRemoverBatchOp()\\\n",
    "            .setSelectedCol(\"tokens\")\\\n",
    "            .setOutputCol(\"left_tokens\")\n",
    "    )\\\n",
    "    .lazyPrint(-1);\n",
    "\n",
    "BatchOperator.execute();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_3\n",
    "\n",
    "titles = getSource()\\\n",
    "    .firstN(10)\\\n",
    "    .select(\"news_title\")\\\n",
    "    .link(\n",
    "        SegmentBatchOp()\\\n",
    "            .setSelectedCol(\"news_title\")\\\n",
    "            .setOutputCol(\"segmented_title\")\\\n",
    "            .setReservedCols(None)\n",
    "    );\n",
    "\n",
    "titles\\\n",
    "    .link(\n",
    "        WordCountBatchOp()\\\n",
    "            .setSelectedCol(\"segmented_title\")\n",
    "    )\\\n",
    "    .orderBy(\"cnt\", 100, order = 'desc')\\\n",
    "    .lazyPrint(-1, \"WordCount\");\n",
    "\n",
    "titles\\\n",
    "    .link(\n",
    "        DocWordCountBatchOp()\\\n",
    "            .setDocIdCol(\"segmented_title\")\\\n",
    "            .setContentCol(\"segmented_title\")\n",
    "    )\\\n",
    "    .lazyPrint(-1, \"DocWordCount\");\n",
    "\n",
    "BatchOperator.execute()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_4\n",
    "titles = getSource()\\\n",
    "    .firstN(10)\\\n",
    "    .select(\"news_title\")\\\n",
    "    .link(\n",
    "        SegmentBatchOp()\\\n",
    "            .setSelectedCol(\"news_title\")\\\n",
    "            .setOutputCol(\"segmented_title\")\\\n",
    "            .setReservedCols(None)\n",
    "    );\n",
    "\n",
    "for featureType in [\"WORD_COUNT\", \"BINARY\", \"TF\", \"IDF\", \"TF_IDF\"] :\n",
    "    DocCountVectorizer()\\\n",
    "        .setFeatureType(featureType)\\\n",
    "        .setSelectedCol(\"segmented_title\")\\\n",
    "        .setOutputCol(\"vec\")\\\n",
    "        .fit(titles)\\\n",
    "        .transform(titles)\\\n",
    "        .lazyPrint(-1, \"DocCountVectorizer + \" + featureType);\n",
    "\n",
    "for featureType in [\"WORD_COUNT\", \"BINARY\", \"TF\", \"IDF\", \"TF_IDF\"] :\n",
    "    DocHashCountVectorizer()\\\n",
    "        .setFeatureType(featureType)\\\n",
    "        .setSelectedCol(\"segmented_title\")\\\n",
    "        .setOutputCol(\"vec\")\\\n",
    "        .setNumFeatures(100)\\\n",
    "        .fit(titles)\\\n",
    "        .transform(titles)\\\n",
    "        .lazyPrint(-1, \"DocHashCountVectorizer + \" + featureType);\n",
    "\n",
    "BatchOperator.execute();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_5_2\n",
    "\n",
    "df = pd.DataFrame([\n",
    "    \"蒸羊羔、蒸熊掌、蒸鹿尾儿、烧花鸭、烧雏鸡、烧子鹅、卤猪、卤鸭、酱鸡、腊肉、松花小肚儿、晾肉、香肠儿、什锦苏盘、熏鸡白肚儿、清蒸八宝猪、江米酿鸭子、罐儿野鸡、罐儿鹌鹑。\"\\\n",
    "    + \"卤什件儿、卤子鹅、山鸡、兔脯、菜蟒、银鱼、清蒸哈什蚂、烩鸭丝、烩鸭腰、烩鸭条、清拌鸭丝、黄心管儿、焖白鳝、焖黄鳝、豆豉鲇鱼、锅烧鲤鱼、烀烂甲鱼、抓炒鲤鱼、抓炒对儿虾。\"\\\n",
    "    + \"软炸里脊、软炸鸡、什锦套肠儿、卤煮寒鸦儿、麻酥油卷儿、熘鲜蘑、熘鱼脯、熘鱼肚、熘鱼片儿、醋熘肉片儿、烩三鲜、烩白蘑、烩鸽子蛋、炒银丝、烩鳗鱼、炒白虾、炝青蛤、炒面鱼。\"\\\n",
    "    + \"炒竹笋、芙蓉燕菜、炒虾仁儿、烩虾仁儿、烩腰花儿、烩海参、炒蹄筋儿、锅烧海参、锅烧白菜、炸木耳、炒肝尖儿、桂花翅子、清蒸翅子、炸飞禽、炸汁儿、炸排骨、清蒸江瑶柱。\"\\\n",
    "    + \"糖熘芡仁米、拌鸡丝、拌肚丝、什锦豆腐、什锦丁儿、糟鸭、糟熘鱼片儿、熘蟹肉、炒蟹肉、烩蟹肉、清拌蟹肉、蒸南瓜、酿倭瓜、炒丝瓜、酿冬瓜、烟鸭掌儿、焖鸭掌儿、焖笋、炝茭白。\"\\\n",
    "    + \"茄子晒炉肉、鸭羹、蟹肉羹、鸡血汤、三鲜木樨汤、红丸子、白丸子、南煎丸子、四喜丸子、三鲜丸子、氽丸子、鲜虾丸子、鱼脯丸子、饹炸丸子、豆腐丸子、樱桃肉、马牙肉、米粉肉。\"\\\n",
    "    + \"一品肉、栗子肉、坛子肉、红焖肉、黄焖肉、酱豆腐肉、晒炉肉、炖肉、黏糊肉、烀肉、扣肉、松肉、罐儿肉、烧肉、大肉、烤肉、白肉、红肘子、白肘子、熏肘子、水晶肘子、蜜蜡肘子。\"\\\n",
    "    + \"锅烧肘子、扒肘条、炖羊肉、酱羊肉、烧羊肉、烤羊肉、清羔羊肉、五香羊肉、氽三样儿、爆三样儿、炸卷果儿、烩散丹、烩酸燕儿、烩银丝、烩白杂碎、氽节子、烩节子、炸绣球。\"\\\n",
    "    + \"三鲜鱼翅、栗子鸡、氽鲤鱼、酱汁鲫鱼、活钻鲤鱼、板鸭、筒子鸡、烩脐肚、烩南荠、爆肚仁儿、盐水肘花儿、锅烧猪蹄儿、拌稂子、炖吊子、烧肝尖儿、烧肥肠儿、烧心、烧肺。\"\\\n",
    "    + \"烧紫盖儿、烧连帖、烧宝盖儿、油炸肺、酱瓜丝儿、山鸡丁儿、拌海蜇、龙须菜、炝冬笋、玉兰片、烧鸳鸯、烧鱼头、烧槟子、烧百合、炸豆腐、炸面筋、炸软巾、糖熘饹儿。\"\\\n",
    "    + \"拔丝山药、糖焖莲子、酿山药、杏仁儿酪、小炒螃蟹、氽大甲、炒荤素儿、什锦葛仙米、鳎目鱼、八代鱼、海鲫鱼、黄花鱼、鲥鱼、带鱼、扒海参、扒燕窝、扒鸡腿儿、扒鸡块儿。\"\\\n",
    "    + \"扒肉、扒面筋、扒三样儿、油泼肉、酱泼肉、炒虾黄、熘蟹黄、炒子蟹、炸子蟹、佛手海参、炸烹儿、炒芡子米、奶汤、翅子汤、三丝汤、熏斑鸠、卤斑鸠、海白米、烩腰丁儿。\"\\\n",
    "    + \"火烧茨菰、炸鹿尾儿、焖鱼头、拌皮渣儿、氽肥肠儿、炸紫盖儿、鸡丝豆苗、十二台菜、汤羊、鹿肉、驼峰、鹿大哈、插根儿、炸花件儿，清拌粉皮儿、炝莴笋、烹芽韭、木樨菜。\"\\\n",
    "    + \"烹丁香、烹大肉、烹白肉、麻辣野鸡、烩酸蕾、熘脊髓、咸肉丝儿、白肉丝儿、荸荠一品锅、素炝春不老、清焖莲子、酸黄菜、烧萝卜、脂油雪花儿菜、烩银耳、炒银枝儿。\"\\\n",
    "    + \"八宝榛子酱、黄鱼锅子、白菜锅子、什锦锅子、汤圆锅子、菊花锅子、杂烩锅子、煮饽饽锅子、肉丁辣酱、炒肉丝、炒肉片儿、烩酸菜、烩白菜、烩豌豆、焖扁豆、氽毛豆、炒豇豆，外加腌苤蓝丝儿。\"\n",
    "])\n",
    "\n",
    "BatchOperator.fromDataframe(df, schemaStr='doc string')\\\n",
    "    .link(\n",
    "        SegmentBatchOp()\\\n",
    "            .setSelectedCol(\"doc\")\\\n",
    "            .setOutputCol(\"words\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StopWordsRemoverBatchOp().setSelectedCol(\"words\")\n",
    "    )\\\n",
    "    .link(\n",
    "        KeywordsExtractionBatchOp()\\\n",
    "            .setMethod('TEXT_RANK')\\\n",
    "            .setSelectedCol(\"words\")\\\n",
    "            .setOutputCol(\"extract_keywords\")\n",
    "    )\\\n",
    "    .select(\"extract_keywords\")\\\n",
    "    .print();\n",
    "\n",
    "getSource()\\\n",
    "    .link(\n",
    "        SegmentBatchOp()\\\n",
    "            .setSelectedCol(\"news_title\")\\\n",
    "            .setOutputCol(\"segmented_title\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StopWordsRemoverBatchOp().setSelectedCol(\"segmented_title\")\n",
    "    )\\\n",
    "    .link(\n",
    "        KeywordsExtractionBatchOp()\\\n",
    "            .setTopN(5)\\\n",
    "            .setMethod('TF_IDF')\\\n",
    "            .setSelectedCol(\"segmented_title\")\\\n",
    "            .setOutputCol(\"extract_keywords\")\n",
    "    )\\\n",
    "    .select(\"news_title, extract_keywords\")\\\n",
    "    .firstN(10)\\\n",
    "    .print();\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_6_1\n",
    "\n",
    "df = pd.DataFrame(\n",
    "    [\n",
    "        [\"机器学习\", \"机器学习\"],\n",
    "        [\"批式计算\", \"流式计算\"],\n",
    "        [\"Machine Learning\", \"ML\"],\n",
    "        [\"Flink\", \"Alink\"],\n",
    "        [\"Good Morning!\", \"Good Evening!\"]\n",
    "    ]\n",
    ")  \n",
    "\n",
    "source = BatchOperator.fromDataframe(df, schemaStr='col1 string, col2 string')\n",
    "\n",
    "source.lazyPrint(-1);\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        StringSimilarityPairwiseBatchOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LEVENSHTEIN\")\\\n",
    "            .setOutputCol(\"LEVENSHTEIN\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StringSimilarityPairwiseBatchOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LEVENSHTEIN_SIM\")\\\n",
    "            .setOutputCol(\"LEVENSHTEIN_SIM\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StringSimilarityPairwiseBatchOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LCS\")\\\n",
    "            .setOutputCol(\"LCS\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StringSimilarityPairwiseBatchOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LCS_SIM\")\\\n",
    "            .setOutputCol(\"LCS_SIM\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StringSimilarityPairwiseBatchOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"JACCARD_SIM\")\\\n",
    "            .setOutputCol(\"JACCARD_SIM\")\n",
    "    )\\\n",
    "    .lazyPrint(-1, \"\\n## StringSimilarityPairwiseBatchOp ##\");\n",
    "\n",
    "source\\\n",
    "    .link(\n",
    "        SegmentBatchOp().setSelectedCol(\"col1\")\n",
    "    )\\\n",
    "    .link(\n",
    "        SegmentBatchOp().setSelectedCol(\"col2\")\n",
    "    )\\\n",
    "    .link(\n",
    "        TextSimilarityPairwiseBatchOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LEVENSHTEIN\")\\\n",
    "            .setOutputCol(\"LEVENSHTEIN\")\n",
    "    )\\\n",
    "    .link(\n",
    "        TextSimilarityPairwiseBatchOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LEVENSHTEIN_SIM\")\\\n",
    "            .setOutputCol(\"LEVENSHTEIN_SIM\")\n",
    "    )\\\n",
    "    .link(\n",
    "        TextSimilarityPairwiseBatchOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LCS\")\\\n",
    "            .setOutputCol(\"LCS\")\n",
    "    )\\\n",
    "    .link(\n",
    "        TextSimilarityPairwiseBatchOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LCS_SIM\")\\\n",
    "            .setOutputCol(\"LCS_SIM\")\n",
    "    )\\\n",
    "    .link(\n",
    "        TextSimilarityPairwiseBatchOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"JACCARD_SIM\")\\\n",
    "            .setOutputCol(\"JACCARD_SIM\")\n",
    "    )\\\n",
    "    .lazyPrint(-1, \"\\n## TextSimilarityPairwiseBatchOp ##\");\n",
    "\n",
    "BatchOperator.execute();\n",
    "\n",
    "\n",
    "source_stream = StreamOperator.fromDataframe(df, schemaStr='col1 string, col2 string')\n",
    "\n",
    "source_stream\\\n",
    "    .link(\n",
    "        StringSimilarityPairwiseStreamOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LEVENSHTEIN\")\\\n",
    "            .setOutputCol(\"LEVENSHTEIN\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StringSimilarityPairwiseStreamOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LEVENSHTEIN_SIM\")\\\n",
    "            .setOutputCol(\"LEVENSHTEIN_SIM\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StringSimilarityPairwiseStreamOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LCS\")\\\n",
    "            .setOutputCol(\"LCS\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StringSimilarityPairwiseStreamOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"LCS_SIM\")\\\n",
    "            .setOutputCol(\"LCS_SIM\")\n",
    "    )\\\n",
    "    .link(\n",
    "        StringSimilarityPairwiseStreamOp()\\\n",
    "            .setSelectedCols([\"col1\", \"col2\"])\\\n",
    "            .setMetric(\"JACCARD_SIM\")\\\n",
    "            .setOutputCol(\"JACCARD_SIM\")\n",
    "    )\\\n",
    "    .print();\n",
    "\n",
    "StreamOperator.execute();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_6_2\n",
    "\n",
    "df = pd.DataFrame(\n",
    "    [\n",
    "        \"林徽因什么理由拒绝了徐志摩而选择梁思成为终身伴侣\",\n",
    "        \"发酵床的垫料种类有哪些？哪种更好？\",\n",
    "        \"京城最值得你来场文化之旅的博物馆\",\n",
    "        \"什么是超写实绘画？\"\n",
    "    ]\n",
    ") \n",
    "target = BatchOperator.fromDataframe(df, schemaStr=TXT_COL_NAME + ' string')\n",
    "\n",
    "source = getSource();\n",
    "\n",
    "for metric in [\"LEVENSHTEIN\", \"LCS\", \"SSK\", \"COSINE\"] :\n",
    "    StringNearestNeighbor()\\\n",
    "        .setMetric(metric)\\\n",
    "        .setSelectedCol(TXT_COL_NAME)\\\n",
    "        .setIdCol(TXT_COL_NAME)\\\n",
    "        .setTopN(5)\\\n",
    "        .setOutputCol(\"similar_titles\")\\\n",
    "        .fit(source)\\\n",
    "        .transform(target)\\\n",
    "        .lazyPrint(-1, \"StringNearestNeighbor + \" + metric);\n",
    "    BatchOperator.execute();\n",
    "\n",
    "\n",
    "for metric in [\"LEVENSHTEIN\", \"LCS\", \"SSK\", \"COSINE\"] :\n",
    "    Pipeline()\\\n",
    "        .add(\n",
    "            Segment()\\\n",
    "                .setSelectedCol(TXT_COL_NAME)\\\n",
    "                .setOutputCol(\"segmented_title\")\n",
    "        )\\\n",
    "        .add(\n",
    "            TextNearestNeighbor()\\\n",
    "                .setMetric(metric)\\\n",
    "                .setSelectedCol(\"segmented_title\")\\\n",
    "                .setIdCol(TXT_COL_NAME)\\\n",
    "                .setTopN(5)\\\n",
    "                .setOutputCol(\"similar_titles\")\n",
    "        )\\\n",
    "        .fit(source)\\\n",
    "        .transform(target)\\\n",
    "        .lazyPrint(-1, \"TextNearestNeighbor + \" + metric);\n",
    "    BatchOperator.execute();\n",
    "\n",
    "\n",
    "for metric in [\"JACCARD_SIM\", \"MINHASH_JACCARD_SIM\", \"SIMHASH_HAMMING_SIM\"] :\n",
    "    StringApproxNearestNeighbor()\\\n",
    "        .setMetric(metric)\\\n",
    "        .setSelectedCol(TXT_COL_NAME)\\\n",
    "        .setIdCol(TXT_COL_NAME)\\\n",
    "        .setTopN(5)\\\n",
    "        .setOutputCol(\"similar_titles\")\\\n",
    "        .fit(source)\\\n",
    "        .transform(target)\\\n",
    "        .lazyPrint(-1, \"StringApproxNearestNeighbor + \" + metric);\n",
    "    BatchOperator.execute();\n",
    "\n",
    "\n",
    "for metric in [\"JACCARD_SIM\", \"MINHASH_JACCARD_SIM\", \"SIMHASH_HAMMING_SIM\"] :\n",
    "    Pipeline()\\\n",
    "        .add(\n",
    "            Segment()\\\n",
    "                .setSelectedCol(TXT_COL_NAME)\\\n",
    "                .setOutputCol(\"segmented_title\")\n",
    "        )\\\n",
    "        .add(\n",
    "            TextApproxNearestNeighbor()\\\n",
    "                .setMetric(metric)\\\n",
    "                .setSelectedCol(\"segmented_title\")\\\n",
    "                .setIdCol(TXT_COL_NAME)\\\n",
    "                .setTopN(5)\\\n",
    "                .setOutputCol(\"similar_titles\")\n",
    "        )\\\n",
    "        .fit(source)\\\n",
    "        .transform(target)\\\n",
    "        .lazyPrint(-1, \"TextApproxNearestNeighbor + \" + metric);\n",
    "    BatchOperator.execute();\n",
    "\n",
    "\n",
    "snn = Pipeline()\\\n",
    "    .add(\n",
    "        StringNearestNeighbor()\\\n",
    "            .setMetric(\"LEVENSHTEIN\")\\\n",
    "            .setSelectedCol(TXT_COL_NAME)\\\n",
    "            .setIdCol(TXT_COL_NAME)\\\n",
    "            .setTopN(5)\\\n",
    "            .setOutputCol(\"similar_titles\")\n",
    "    );\n",
    "\n",
    "approx_snn = Pipeline()\\\n",
    "    .add(\n",
    "        StringApproxNearestNeighbor()\\\n",
    "            .setMetric(\"JACCARD_SIM\")\\\n",
    "            .setSelectedCol(TXT_COL_NAME)\\\n",
    "            .setIdCol(TXT_COL_NAME)\\\n",
    "            .setTopN(5)\\\n",
    "            .setOutputCol(\"similar_titles\")\n",
    "    );\n",
    "\n",
    "sw = Stopwatch();\n",
    "\n",
    "if not(os.path.exists(DATA_DIR + SNN_MODEL_FILE)) :\n",
    "    sw.reset();\n",
    "    sw.start();\n",
    "    snn.fit(source).save(DATA_DIR + SNN_MODEL_FILE);\n",
    "    BatchOperator.execute();\n",
    "    sw.stop();\n",
    "    print(sw.getElapsedTimeSpan());\n",
    "\n",
    "\n",
    "if not(os.path.exists(DATA_DIR + APPROX_SNN_MODEL_FILE)) :\n",
    "    sw.reset();\n",
    "    sw.start();\n",
    "    approx_snn.fit(source).save(DATA_DIR + APPROX_SNN_MODEL_FILE);\n",
    "    BatchOperator.execute();\n",
    "    sw.stop();\n",
    "    print(sw.getElapsedTimeSpan());\n",
    "\n",
    "\n",
    "target_stock = source.filter(\"category_name = 'stock'\");\n",
    "target_news_story = source.filter(\"category_name = 'news_story'\");\n",
    "\n",
    "sw.reset();\n",
    "sw.start();\n",
    "PipelineModel\\\n",
    "    .load(DATA_DIR + SNN_MODEL_FILE)\\\n",
    "    .transform(target_stock)\\\n",
    "    .lazyPrint(10, \"StringNeareastNeighbor + LEVENSHTEIN\");\n",
    "BatchOperator.execute();\n",
    "sw.stop();\n",
    "print(sw.getElapsedTimeSpan());\n",
    "\n",
    "sw.reset();\n",
    "sw.start();\n",
    "PipelineModel\\\n",
    "    .load(DATA_DIR + APPROX_SNN_MODEL_FILE)\\\n",
    "    .transform(target_stock)\\\n",
    "    .lazyPrint(10, \"JACCARD_SIM + stock\");\n",
    "BatchOperator.execute();\n",
    "sw.stop();\n",
    "print(sw.getElapsedTimeSpan());\n",
    "\n",
    "sw.reset();\n",
    "sw.start();\n",
    "PipelineModel\\\n",
    "    .load(DATA_DIR + APPROX_SNN_MODEL_FILE)\\\n",
    "    .transform(target_news_story)\\\n",
    "    .lazyPrint(10, \"JACCARD_SIM + news_story\");\n",
    "BatchOperator.execute();\n",
    "sw.stop();\n",
    "print(sw.getElapsedTimeSpan());\n",
    "\n",
    "#StreamOperator.setParallelism(1);\n",
    "\n",
    "stream_target = StreamOperator.fromDataframe(df, schemaStr=TXT_COL_NAME + ' string')\n",
    "\n",
    "PipelineModel\\\n",
    "    .load(DATA_DIR + SNN_MODEL_FILE)\\\n",
    "    .transform(stream_target)\\\n",
    "    .print();\n",
    "StreamOperator.execute();\n",
    "\n",
    "stream_target_stock = getStreamSource().filter(\"category_name = 'stock'\");\n",
    "\n",
    "sw.reset();\n",
    "sw.start();\n",
    "PipelineModel\\\n",
    "    .load(DATA_DIR + APPROX_SNN_MODEL_FILE)\\\n",
    "    .transform(stream_target_stock)\\\n",
    "    .sample(0.02)\\\n",
    "    .print();\n",
    "StreamOperator.execute();\n",
    "sw.stop();\n",
    "print(sw.getElapsedTimeSpan());"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_7\n",
    "\n",
    "docs = getSource()\\\n",
    "    .select(LABEL_COL_NAME + \", \" + TXT_COL_NAME)\\\n",
    "    .link(SegmentBatchOp().setSelectedCol(TXT_COL_NAME))\\\n",
    "    .link(StopWordsRemoverBatchOp().setSelectedCol(TXT_COL_NAME));\n",
    "\n",
    "docs.lazyPrint(10);\n",
    "\n",
    "if not(os.path.exists(DATA_DIR + LDA_MODEL_FILE)) :\n",
    "    lda = LdaTrainBatchOp()\\\n",
    "        .setTopicNum(10)\\\n",
    "        .setNumIter(200)\\\n",
    "        .setVocabSize(20000)\\\n",
    "        .setSelectedCol(TXT_COL_NAME)\\\n",
    "        .setRandomSeed(123);\n",
    "\n",
    "    docs.link(lda);\n",
    "\n",
    "    lda.lazyPrintModelInfo();\n",
    "\n",
    "    lda.link(AkSinkBatchOp().setFilePath(DATA_DIR + LDA_MODEL_FILE));\n",
    "\n",
    "    lda.getSideOutput(0).link(AkSinkBatchOp().setFilePath(DATA_DIR + LDA_PWZ_FILE));\n",
    "\n",
    "    BatchOperator.execute();\n",
    "\n",
    "\n",
    "LdaPredictBatchOp()\\\n",
    "    .setSelectedCol(TXT_COL_NAME)\\\n",
    "    .setPredictionCol(PREDICTION_COL_NAME)\\\n",
    "    .setPredictionDetailCol(\"predinfo\")\\\n",
    "    .linkFrom(\n",
    "        AkSourceBatchOp().setFilePath(DATA_DIR + LDA_MODEL_FILE),\n",
    "        docs\n",
    "    )\\\n",
    "    .lazyPrint(5)\\\n",
    "    .link(\n",
    "        EvalClusterBatchOp()\\\n",
    "            .setLabelCol(LABEL_COL_NAME)\\\n",
    "            .setPredictionCol(PREDICTION_COL_NAME)\\\n",
    "            .lazyPrintMetrics()\n",
    "    );\n",
    "\n",
    "pwz = AkSourceBatchOp().setFilePath(DATA_DIR + LDA_PWZ_FILE);\n",
    "\n",
    "pwz.sample(0.001).lazyPrint(10);\n",
    "\n",
    "for t in range(0, 10) :\n",
    "    pwz.select(\"word, topic_\" + str(t))\\\n",
    "        .orderBy(\"topic_\" + str(t), 20, order = 'desc')\\\n",
    "        .lazyPrint(-1, \"topic\" + str(t));\n",
    "\n",
    "BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
